package rxgo

import (
	"context"
	"reflect"
	"sync"
)

type mOper struct {
	opFunc func(ctx context.Context, o *Observable, item reflect.Value, out chan interface{}) (end bool)
}

func (tsop mOper) op(ctx context.Context, o *Observable) {
	in := o.pred.outflow
	out := o.outflow
	var waitG sync.WaitGroup

	go func() {
		end := false
		for x := range in {
			if end {
				continue
			}
			xv := reflect.ValueOf(x)
			if e, ok := x.(error); ok && !o.flip_accept_error { //错误检测
				o.sendToFlow(ctx, e, out)
				continue
			}
			switch threading := o.threading; threading {
			case ThreadingDefault:
				if tsop.opFunc(ctx, o, xv, out) {
					end = true
				}
			case ThreadingIO:
				fallthrough
			case ThreadingComputing:
				waitG.Add(1)
				go func() {
					defer waitG.Done()
					if tsop.opFunc(ctx, o, xv, out) {
						end = true
					}
				}()
			default:
			}
		}

		waitG.Wait()
		o.sendToFlow(ctx, o.dat, out)
		o.closeFlow(out)
	}()
}

func (parent *Observable) MObservable(name string) (o *Observable) {
	o = newObservable()
	o.Name = name

	parent.next = o
	o.pred = parent
	o.root = parent.root

	o.buf_len = BufferLen
	return o
}

var MOper = mOper{func(ctx context.Context, o *Observable, x reflect.Value, out chan interface{}) (end bool) {
	o.mathFunc(o, int(x.Int()))
	return}
}

func MSum(o *Observable, x int) { //求和
	o.dat += x
}

func MCnt(o *Observable, x int) { //计数
	o.dat += 1
}

func MAvg(o *Observable, x int) { //求平均值
	o.sum += x
	o.cnt += 1
	o.dat = o.sum / o.cnt
}

func MMin(o *Observable, x int) { //求最小值
	o.cnt += 1
	if o.cnt == 1 || x < o.dat {
		o.dat = x
	}
}

func MMax(o *Observable, x int) { //求最大值
	o.cnt += 1
	if o.cnt == 1 || x > o.dat {
		o.dat = x
	}
}

func (parent *Observable) Sum() (o *Observable) { //求和函数
	o = parent.MObservable("Sum")
	o.sum = 0
	o.cnt = 0
	o.dat = 0
	o.operator = MOper
	o.mathFunc = MSum
	return o
}

func (parent *Observable) Count() (o *Observable) { //计数函数
	o = parent.MObservable("Count")
	o.sum = 0
	o.cnt = 0
	o.dat = 0
	o.operator = MOper
	o.mathFunc = MCnt
	return o
}

func (parent *Observable) Average() (o *Observable) { //均值函数
	o = parent.MObservable("Average")
	o.sum = 0
	o.cnt = 0
	o.dat = 0
	o.operator = MOper
	o.mathFunc = MAvg
	return o
}

func (parent *Observable) Min() (o *Observable) { //最小值函数
	o = parent.MObservable("Min")
	o.sum = 0
	o.cnt = 0
	o.dat = 0
	o.operator = MOper
	o.mathFunc = MMin
	return o
}

func (parent *Observable) Max() (o *Observable) { //最大值函数
	o = parent.MObservable("Max")
	o.sum = 0
	o.cnt = 0
	o.dat = 0
	o.operator = MOper
	o.mathFunc = MMax
	return o
}
